package gnss

import (
	"dgo/goutils/gis"
	"dgo/goutils/utils"
	"encoding/base64"
	"fmt"
	"golang.org/x/net/proxy"
	"io"
	"net"
	"runtime"
	"strings"
	"sync"
	"sync/atomic"
	"time"
)

type NtripClientCltV2023 struct {
	disflag   int32
	stopflag  int32
	disreason string
	lk        sync.RWMutex

	closewg sync.WaitGroup

	authstate    int8
	execflag     int8
	startedflag  int8
	responseAuth string

	lastDialAddr    string
	dialProxyAddr   string
	addr            string
	user, pass      string
	userAgent       string
	requestService  string
	conn            net.Conn
	requestRawHead  string
	lastConnectTime time.Time

	// 最小重连间隔
	reconnectMinInterval time.Duration

	OnBeforeWriteBuf   func(sender *NtripClientCltV2023, buf []byte) (newBuf []byte)
	OnWriteBuf         func(sender *NtripClientCltV2023, buf []byte)
	OnBeforeConnect    func(sender *NtripClientCltV2023, lastconnect time.Time) bool
	OnConnectFail      func(sender *NtripClientCltV2023, err error)
	OnConnected        func(sender *NtripClientCltV2023)
	OnBuildRequestHead func(sender *NtripClientCltV2023, hb *utils.BytesBuilder)
	OnResponseAuth     func(sender *NtripClientCltV2023, resp string, err error)

	// buf 可以进行异步
	OnRecvBuf      func(sender *NtripClientCltV2023, buf []byte)
	OnDisconnected func(sender *NtripClientCltV2023, reason string)

	OnMakeRecvBlock func() []byte
}

var (
	ntripclt2023_obj_cnt  int32
	ntripclt2023_work_cnt int32
)

func GetNtripClientCltV2023ObjAliveCnt() int32 {
	return atomic.LoadInt32(&ntripclt2023_obj_cnt)
}

func GetNtripClientCltV2023ObjStatus() string {
	return fmt.Sprintf("alive:%d, work:%d", ntripclt2023_obj_cnt, ntripclt2023_work_cnt)
}

func NewNtripClientCltV2023() *NtripClientCltV2023 {
	rval := &NtripClientCltV2023{
		reconnectMinInterval: time.Second * 5,
	}
	atomic.AddInt32(&ntripclt2023_obj_cnt, 1)
	runtime.SetFinalizer(rval, func(obj interface{}) {
		atomic.AddInt32(&ntripclt2023_obj_cnt, -1)
	})

	return rval
}

/*
1: 认证成功
-1： 认证失败
*/
func (this *NtripClientCltV2023) AuthState() int {
	return int(this.authstate)
}

// 是否已经开启(进入了接收线程)
func (this *NtripClientCltV2023) StartedFlag() int8 {
	return this.startedflag
}

func (this *NtripClientCltV2023) GetUser() string {
	return this.user
}

func (this *NtripClientCltV2023) GetUserAgent() string {
	return this.userAgent
}

func (this *NtripClientCltV2023) GetRequestService() string {
	return this.requestService
}

func (this *NtripClientCltV2023) GetRemoteAddr() string {
	return this.addr
}

func (this *NtripClientCltV2023) GetLatestRequestRawHead() string {
	return this.requestRawHead
}

/*
配置代理拨号地址
*/
func (this *NtripClientCltV2023) ConfigDialAddr(s string) bool {
	if this.dialProxyAddr != s {
		this.dialProxyAddr = s
		return true
	}
	return false
}

func (this *NtripClientCltV2023) ConfigConnectString(s string) bool {
	conf := ParseNtripClientConnectString(s)
	ischanged := this.ConfigRemoteAddr(conf.Addr)
	ischanged = this.ConfigUserPass(conf.User, conf.Pass) || ischanged
	ischanged = this.ConfigRequestService(conf.MountPoint) || ischanged
	return ischanged
}

func (this *NtripClientCltV2023) ConfigRemoteAddr(s string) bool {
	if this.addr != s {
		this.addr = s
		return true
	}
	return false
}

func (this *NtripClientCltV2023) ConfigReconnectMinInterval(interval time.Duration) {
	this.reconnectMinInterval = interval
	if this.reconnectMinInterval < time.Millisecond*100 {
		this.reconnectMinInterval = time.Second
	}
}

func (this *NtripClientCltV2023) ConfigUserAgent(s string) bool {
	if this.userAgent != s {
		this.userAgent = s
		return true
	}
	return false
}

func (this *NtripClientCltV2023) ConnString(mixPass bool) string {
	if mixPass {
		return fmt.Sprintf("%s:***@%s/%s", this.user, this.lastDialAddr, this.requestService)
	} else {
		return fmt.Sprintf("%s:%s@%s/%s", this.user, this.pass, this.lastDialAddr, this.requestService)
	}
}

func (this *NtripClientCltV2023) ConfigUserPass(user, pass string) (ischanged bool) {
	if this.user != user {
		this.user = user
		ischanged = true
	}
	if this.pass != pass {
		this.pass = pass
		ischanged = true
	}
	return
}

func (this *NtripClientCltV2023) WriteBuf(buf []byte) error {
	conn := this.conn
	if conn == nil {
		return io.ErrClosedPipe
	}
	newBuf := this.doBeforeWriteBuf(buf)
	if len(newBuf) == 0 {
		return nil
	}
	n, err := conn.Write(newBuf)
	if err != nil {
		this.RequestDis(fmt.Sprintf("write err:%s", err.Error()))
	} else {
		this.doWriteBuf(newBuf[:n])
	}
	return err
}

func (this *NtripClientCltV2023) ConfigRequestService(s string) (ischanged bool) {
	if this.requestService != s {
		this.requestService = s
		ischanged = true
	}
	return
}

func (this *NtripClientCltV2023) PostNmeaTempl(nmeaTempl string) string {
	if utils.GoFunCatchException {
		defer utils.PanicHandler()
	}
	nmea := gis.NewNMEA()
	nmea.ParseText(utils.Trim(nmeaTempl))
	nmea.UpdateUTCTime()
	nmeaBuf := nmea.EncodeBuf()
	this.WriteBuf(nmeaBuf)
	return string(nmeaBuf)
}

func (this *NtripClientCltV2023) WriteNmeaTempl(nmeaTempl string) (err error, sendNmea string) {
	nmea := gis.NewNMEA()
	nmea.ParseText(utils.Trim(nmeaTempl))
	nmea.UpdateUTCTime()
	nmeaBuf := nmea.EncodeBuf()
	err = this.WriteBuf(nmeaBuf)
	sendNmea = string(nmeaBuf)
	return
}

func (this *NtripClientCltV2023) Connected() bool {
	return this.conn != nil && atomic.LoadInt32(&this.disflag) == 0
}

func (this *NtripClientCltV2023) doConnectFail(err error) {
	evt := this.OnConnectFail
	if evt != nil {
		evt(this, err)
	}
}

func (this *NtripClientCltV2023) doDisconnected() {
	evt := this.OnDisconnected
	if evt != nil {
		evt(this, this.disreason)
	}
}

func (this *NtripClientCltV2023) doRecvBuf(buf []byte) {
	evt := this.OnRecvBuf
	if evt != nil {
		evt(this, buf)
	}
}

func (this *NtripClientCltV2023) doBeforeWriteBuf(buf []byte) (newBuf []byte) {
	evt := this.OnBeforeWriteBuf
	if evt != nil {
		return evt(this, buf)
	}
	return buf
}

func (this *NtripClientCltV2023) doWriteBuf(buf []byte) {
	evt := this.OnWriteBuf
	if evt != nil {
		evt(this, buf)
	}
}

func (this *NtripClientCltV2023) doConnected() {
	evt := this.OnConnected
	if evt != nil {
		evt(this)
	}
}

func (this *NtripClientCltV2023) doBeforeConnect() bool {
	if time.Since(this.lastConnectTime) >= this.reconnectMinInterval {
		evt := this.OnBeforeConnect
		if evt != nil {
			return evt(this, this.lastConnectTime)
		}
		return true
	} else { // 重连时间过短
		return false
	}
}

func (this *NtripClientCltV2023) doResponseAuth(err error) {
	evt := this.OnResponseAuth
	if evt != nil {
		evt(this, this.responseAuth, err)
	}
}

func (this *NtripClientCltV2023) writeRequestBuf(user, pass, mountpoint, useragent string) error {
	conn := this.conn
	if conn == nil {
		return io.ErrClosedPipe
	}
	var sb utils.BytesBuilder
	sb.Appendf("GET /%s HTTP/1.0\r\n", mountpoint)
	if len(useragent) > 0 {
		sb.Appendf("User-Agent: NTRIP %s\r\n", useragent)
	} else {
		sb.Appendf("User-Agent: NTRIP %s\r\n", "DIOCPNtripClient")
	}
	if len(user) > 0 {
		str := user + ":" + pass
		authstr := base64.StdEncoding.EncodeToString([]byte(str))
		sb.Appendf("Authorization: Basic %s\r\n", authstr)
	}
	evt := this.OnBuildRequestHead
	if evt != nil {
		evt(this, &sb)
	}
	sb.AppendStr("\r\n")
	str := sb.String()
	this.requestRawHead = str
	return this.WriteBuf([]byte(str))
}

func (this *NtripClientCltV2023) connect() (err error) {
	this.lastConnectTime = time.Now()
	proxyaddr := this.dialProxyAddr
	if len(proxyaddr) > 0 {
		this.lastDialAddr = fmt.Sprintf("%s,%s", proxyaddr, this.addr)
		d, err := proxy.SOCKS5("tcp", proxyaddr, nil, proxy.Direct)
		if err != nil {
			return err
		}
		conn, err := d.Dial("tcp", this.addr)
		if err != nil {
			return err
		}
		this.conn = conn
	} else {
		this.lastDialAddr = fmt.Sprintf("%s", this.addr)
		conn, err := net.Dial("tcp", this.addr)
		if err != nil {
			return err
		}
		this.conn = conn
	}

	this.authstate = 0
	atomic.StoreInt32(&this.disflag, 0)
	return nil
}

func (this *NtripClientCltV2023) GetDisReason() string {
	return this.disreason
}

func (this *NtripClientCltV2023) RequestDis(reason string) {
	if atomic.CompareAndSwapInt32(&this.disflag, 0, 1) {
		this.disreason = reason
	}
	conn := this.conn
	if conn != nil {
		conn.Close()
	}
}

/*
请求停止, 必须在start之后掉用， 否则无效
*/
func (this *NtripClientCltV2023) RequestStop(reason string) {
	atomic.StoreInt32(&this.stopflag, 1)
	this.RequestDis(reason)
}

func (this *NtripClientCltV2023) Stop() {
	this.RequestStop("请求关闭(Stop)")
}

/*
1.重置(新建的不需要调用)
2.开启/Stop

完全用完后可以调用, 这样可以进行回收
*/
func (this *NtripClientCltV2023) Reset() {
	atomic.StoreInt32(&this.stopflag, 0)
	atomic.StoreInt32(&this.disflag, 0)
	this.conn = nil
	this.CleanEvent()
	this.dialProxyAddr = ""
	this.lastDialAddr = ""
	this.responseAuth = ""
	this.requestService = ""
	this.requestRawHead = ""
	this.addr = ""
	this.user = ""
	this.pass = ""
	this.userAgent = ""
	this.disreason = ""
	this.startedflag = 0
}

func (this *NtripClientCltV2023) CleanEvent() {
	this.OnMakeRecvBlock = nil
	this.OnConnected = nil
	this.OnBuildRequestHead = nil
	this.OnBeforeConnect = nil
	this.OnDisconnected = nil
	this.OnConnectFail = nil
	this.OnResponseAuth = nil
	this.OnRecvBuf = nil
	this.OnWriteBuf = nil
	this.OnBeforeWriteBuf = nil
}

/*
不要在回调事件中回调
*/
func (this *NtripClientCltV2023) StopAndWait() {
	this.Stop()
	this.closewg.Wait()
}

/*
如果认证成功：

			first返回认证成功信息
		    cache中是其他收到的数据
	        err 返回nil;

否则：

	first 返回第一次收到的所有数据, 直到关闭或者数据大于>1024
*/
func NtripCliReadFirst(r io.Reader, cache *utils.BytesBuilder) (authstate int8, first string, err error) {
	conn, _ := r.(net.Conn)
	if conn != nil {
		conn.SetReadDeadline(time.Now().Add(time.Second * 10))
		defer conn.SetReadDeadline(utils.ZeroTime)
	}
	buf := make([]byte, 256)
	chkflag := 0
	for {
		n, e0 := r.Read(buf)
		if e0 != nil {
			if e0 == io.EOF {
				first = cache.String()
				if len(first) == 0 {
					first = "等待响应超时"
				}
				return
			} else {
				first = cache.String()
				err = e0
				return
			}
		}
		cache.Write(buf[:n])
		if chkflag == 0 {
			idx := cache.Index(0, []byte("\r\n"))
			if idx == -1 { // 没有换行符
				if cache.Len() > 512 { // 512都没有换行符
					first = cache.String()
					return
				}
				continue
			}
			str := strings.ToUpper(utils.Trim(string(cache.Buff(0, idx))))
			if strings.Contains(str, "ICY 200 OK") {
				first = string(cache.Buff(0, idx+2))
				cache.Delete(0, idx+2)
				for {
					if cache.Len() > 0 && (cache.Byte(0) == '\r' || cache.Byte(0) == '\n') {
						cache.Delete(0, 1)
					} else {
						break
					}
				}
				authstate = 1
				return
			} else if strings.Contains(str, "UNAUTHORIZED") {
				authstate = -1
				chkflag = 1
			} else if strings.Contains(str, "SOURCETABLE") { // SOURCETABLE 200 OK
				chkflag = 1
				authstate = 0
			} else if strings.Index(str, "ERROR ") == 0 { //
				chkflag = 1
				authstate = -1
			}
		}
	}
}

func (this *NtripClientCltV2023) CheckStart() bool {
	if this.execflag != 1 {
		go this.Start()
		time.Sleep(time.Millisecond * 100)
		return true
	} else {
		return false
	}
}

func (this *NtripClientCltV2023) Started() bool {
	return this.execflag == 1
}

func (this *NtripClientCltV2023) Start() {
	if utils.GoFunCatchException {
		defer utils.PanicHandler()
	}
	atomic.AddInt32(&ntripclt2023_work_cnt, 1)
	this.execflag = 1
	this.closewg.Add(1)
	defer func() {
		this.execflag = 0
		this.closewg.Done()
		atomic.AddInt32(&ntripclt2023_work_cnt, -1)
	}()

	makeRecvBlock := this.OnMakeRecvBlock
	if makeRecvBlock == nil {
		makeRecvBlock = func() []byte {
			return make([]byte, 1024)
		}
	}

	this.startedflag = 1
	for {

		if atomic.LoadInt32(&this.stopflag) == 1 {
			break
		}

		if !this.Connected() {
			if !this.doBeforeConnect() {
				time.Sleep(time.Millisecond * 100)
				continue
			}

			err := this.connect()
			if err != nil {
				this.doConnectFail(err)
				continue
			}
			this.doConnected()
			if atomic.LoadInt32(&this.stopflag) == 1 {
				this.RequestDis("stop!!!")
				this.doDisconnected()
				continue
			}
			err = this.writeRequestBuf(this.user, this.pass, this.requestService, this.userAgent)
			if err != nil {
				this.RequestDis(fmt.Sprintf("write request head err:%s", err.Error()))
				this.doDisconnected()
				continue
			}
			var cache utils.BytesBuilder

			this.authstate, this.responseAuth, err = NtripCliReadFirst(this.conn, &cache)
			if this.authstate != 1 {
				this.doResponseAuth(err)
				this.RequestDis(fmt.Sprintf("%s", this.responseAuth))
				this.doDisconnected()
				continue
			} else if err != nil {
				this.responseAuth = fmt.Sprintf("read response err:%s", err.Error())
				this.doResponseAuth(err)
				this.RequestDis(fmt.Sprintf("read first err:%s", err.Error()))
				this.doDisconnected()
				continue
			}
			//this.conn.SetReadDeadline(utils.ZeroTime)
			// 认证成功
			this.doResponseAuth(nil)
			if cache.Len() > 0 {
				this.doRecvBuf(cache.Bytes())
			}
		}

		buf := makeRecvBlock() // 方便进行异步处理
		n, err := this.conn.Read(buf)
		if err != nil {
			this.RequestDis(fmt.Sprintf("接收数据出现异常:%s", err.Error()))
			this.doDisconnected()
			continue
		}
		this.doRecvBuf(buf[:n])
	}
}
